package ec_fourthinfo;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class FourthInfoV_1 {
	private static String[] fields = { "stat_date", "dispcity1", "dispcity2",
			"dispcity3", "belong_cate1", "belong_cate2", "total_info_count",
			"total_info_users", "valid_info_count", "valid_info_users",
			"new_info_count", "new_info_users", "modify_info_count",
			"modify_info_users", "refresh_info_count", "refresh_info_users",
			"new_valid_info_count", "new_valid_info_users",
			"ordinary_valid_info_count", "ordinary_valid_info_users",
			"ordinary_new_info_count", "ordinary_new_info_users",
			"ordinary_modify_info_count", "ordinary_modify_info_users",
			"ordinary_refresh_info_count", "ordinary_refresh_info_users",
			"ordinary_new_valid_info_count", "ordinary_new_valid_info_users",
			"vip_valid_info_count", "vip_valid_info_users",
			"vip_new_info_count", "vip_new_info_users",
			"vip_modify_info_count", "vip_modify_info_users",
			"vip_refresh_info_count", "vip_refresh_info_users",
			"vip_new_valid_info_count", "vip_new_valid_info_users" };

	private static HashMap<String, Integer> fieldIndex = null;
	private static HashMap<String, String> dict = new HashMap<String, String>();
	private static HashMap<String, String> channelInfo = null;
	private static HashMap<String, ArrayList<String>> cateinfo = null;
	private static HashMap<String, ArrayList<String>> localinfo = null;

	// Mapper
	public static class FourthInfoMapper extends Mapper<LongWritable, Text, Text, Text> {
		private static Pattern p = Pattern.compile("^cate(first|second)(\\d+)_local(first|second|thirt)(\\d+)_(.*?)(\\d{8})_(.*)");
		private Text mapOutputKey = new Text();
		private Text mapOutputValue = new Text();

		public void map(LongWritable key, Text value, Context context) {
			Matcher m = p.matcher(value.toString());
			if (m.find()) {
				String cateid_localid = m.group(2) + "_" + m.group(4);
				String info = m.group(5) + m.group(7);
				mapOutputKey.set(cateid_localid);
				mapOutputValue.set(info);
				try {
					context.write(mapOutputKey, mapOutputValue);
				} catch (Exception e) {
					e.printStackTrace();
					return;
				}
			}
		}
	}

	// Reducer
	public static class FourthInfoReducer extends Reducer<Text, Text, NullWritable, Text> {
		private MultipleOutputs<NullWritable, Text> mos;
		private String date;
		private NullWritable nullKey=NullWritable.get();
		private Text outputValue=new Text();
		public void setup(Context context) {
			mos = new MultipleOutputs<NullWritable, Text>(context);
			date = context.getConfiguration().get("DATE");
		}

		public void cleanup(Context context) throws IOException {
			try {
				mos.close();
			} catch (InterruptedException e) {
				e.printStackTrace();
				return;
			}
		}

		public void reduce(Text key, Iterable<Text> values, Context context) {
			String[] cate_local_id=key.toString().split("_");
			String cateid=cate_local_id[0]; 
			String localid=cate_local_id[1];
			ArrayList<String> cate=cateinfo.get(cateid);
			ArrayList<String> local=localinfo.get(localid);
			String[] row=new String[fields.length];
			row[0]=date;
			int index=fieldIndex.get("belong_cate1");
			row[index]=cate.get(0);
			row[index+1]=cate.get(1);
			index=fieldIndex.get("dispcity1");
			for(int i=0;i<3;i++){
				row[index+i]=local.get(i);
			}
			for(Text t:values){
				String[] info=t.toString().split(":");
				String infoKey=info[0];
				String infoValue=info[1];
				int fieldindex=fieldIndex.get(dict.get(infoKey));
				row[fieldindex]=infoValue;
			}
			
			String tablename=channelInfo.get(cate.get(2));
			 try {
				mos.write("multiple",nullKey, outputValue, date+"/"+tablename+"/part");
			} catch (Exception e) {
				e.printStackTrace();
				return;
			}
		}
	}

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		init();
		Configuration conf = new Configuration();
		conf.set("mapred.job.queue.name", "regular");
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 1) {
			System.out
					.println("************************************************");
			System.out
					.println("************************************************");
			System.out
					.println("Usage: please input 1 params, for example: file.jar args[0]");
			System.out
					.println("args[0] is dateList: 20130101,20130106 or 20130106");
			System.out
					.println("************************************************");
			System.out
					.println("************************************************");
			System.exit(127);
		}
		int startDate = Integer.parseInt(otherArgs[0].split(",")[0]);
		int endDate =otherArgs[0].split(",").length == 2 ? Integer.parseInt(otherArgs[0].split(",")[1]): startDate;
		String baseInPath = "/dsap/rawdata/track.58.com";
		String baseOutPath = "/dsap/middata/zhanghao06/TrackinfoDemo";
		int returnCode = 126;

		for (int i=startDate;i<endDate;i++) {
			conf.set("DATE", String.valueOf(i));
			Job job = new Job(conf, "Fourthinfo");
			job.setJarByClass(FourthInfoV_1.class);
			job.setMapperClass(FourthInfoMapper.class);
			job.setReducerClass(FourthInfoReducer.class);
			job.setNumReduceTasks(11);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			MultipleOutputs.addNamedOutput(job, "multiple", TextOutputFormat.class,
					 NullWritable.class, Text.class);
			if (FileSystem.get(conf).exists(new Path(baseInPath + "/" + i))) {
				FileInputFormat.addInputPath(job, new Path(baseInPath + "/" + i));
			} else {
				continue;
			}
			String outPath = baseOutPath + "/" + i;
			FileSystem.get(conf).delete(new Path(outPath), true);
			FileOutputFormat.setOutputPath(job, new Path(outPath));
			returnCode = job.waitForCompletion(true) ? 0 : 1;
		}

		System.exit(returnCode);

	}

	public static void init() throws SQLException {
		String[] states = { "state1", "uptype0", "uptype1", "uptype2",
				"state1_uptype0" };
		String[] types = { "valid", "new", "modify", "refresh", "new_valid" };

		String[] src = { "", "source0", "source6" };
		String[] roles = { "", "ordinary", "vip" };

		String[] countType = { "_info_count", "_info_users" };
		String[] counters = { "icount", "ucount" };
		for (int i = 0; i < types.length; i++)
			for (int j = 0; j < roles.length; j++)
				for (int k = 0; k < countType.length; k++) {
					if(j==0)
						dict.put(states[i] + "_" + counters[k], types[i] + "_" + countType[k]);
					else
						dict.put(states[i] + "_" + src[j] + "_" + counters[k],roles[j] + "_" + types[i] + "_" + countType[k]);
				}
		dict.put("icount", "total_info_count");
		dict.put("ucount", "total_info_users");
		fieldIndex = new HashMap<String, Integer>();
		for (int i = 0; i < fields.length; i++) {
			fieldIndex.put(fields[i], i);
		}
		Connection conn = getConnection();
		Statement stmt = conn.createStatement();
		String sql = "select cate_id, cate1, cate2, channel_id from belong_cate where cate3_id=0;";
		cateinfo = getInfo(stmt, sql);
		System.out.println(cateinfo.size()+" cateinfo!");
		sql = "select local_id, city1, city2, city3 from display_local where city4_id=0;";
		localinfo = getInfo(stmt, sql);
		System.out.println(localinfo.size()+" localinfo!");
		sql = "select channel_id, table_prefix from channel_define;";
		channelInfo = getchannelInfo(stmt, sql);
		System.out.println(channelInfo.size()+" channelInfo!");
		if (stmt != null)
			stmt.close();
		if (conn != null)
			conn.close();
		stmt = null;
		conn = null;
	}

	private static HashMap<String, ArrayList<String>> getInfo(Statement stmt, String sql) {
		HashMap<String, ArrayList<String>> info = new HashMap<String, ArrayList<String>>();
		try {
			ResultSet rs = stmt.executeQuery(sql);
			int columnCount = rs.getMetaData().getColumnCount();
			while (rs.next()) {
				String id = rs.getString(1);
				ArrayList<String> columns = new ArrayList<String>();
				for (int i = 2; i <= columnCount; i++)
					columns.add(rs.getString(i));
				info.put(id, columns);
			}
			if (rs != null)
				rs.close();
			rs = null;
		} catch (SQLException e) {
			e.printStackTrace();
			return info;
		}
		return info;
	}

	private static HashMap<String, String> getchannelInfo(Statement stmt,
			String sql) {
		HashMap<String, String> info = new HashMap<String, String>();
		try {
			ResultSet rs = stmt.executeQuery(sql);
			int columnCount = rs.getMetaData().getColumnCount();
			while (rs.next()) {
				info.put(rs.getString(1), rs.getString(2));
			}
			if (rs != null)
				rs.close();
			rs = null;
		} catch (SQLException e) {
			e.printStackTrace();
			return info;
		}
		return info;
	}

	private static Connection getConnection() {
		Connection conn = null;
		try {

			InputStream ins = FourthInfoV_1.class.getClassLoader()
					.getResourceAsStream("db.properties");
			Properties p = new Properties();
			p.load(ins);
			ins.close();
			Class.forName(p.getProperty("DriversClass"));
			String url = p.getProperty("ConnetionURL");
			String user = p.getProperty("UserName");
			String password = p.getProperty("PassWord");
			conn = DriverManager.getConnection(url, user, password);
			// log.debug("DbPool: 创建数据库连接池：" + name);
		} catch (Exception e) {
			e.printStackTrace();
			// log.error("DbPool: 创建数据源 " + name + " 失败：",e);
			return null;
		}
		return conn;
	}

}
